//=========================================================================
// Copyright (C) 2025 The C++ Component Model(COMO) Open Source Project
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//      http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//=========================================================================

#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <assert.h>
#include <sys/select.h>
#include <unistd.h>
#include <math.h>
#include <time.h>
#include <syslog.h>
#include "zmtp.h"
#include "int64.h"
#include "lamport_clock.h"
#include "math_utils.h"
#include "checksum.h"
#include "voter_message.h"
#include "voter_lib.h"
#include "voter_server.h"
#include "ZMQ_Switcher.h"
#include "voter_logger.h"
#include "config.h"
#include "rudp.h"
#include "voter_customized.h"
#ifdef SYNC_WITH_PHXPAXOS
  #include "PhxPaxosC.h"
#endif
#include "voter_VotingAlgorithms.h"

#define max(a,b) ((a) >= (b) ? (a) : (b))

extern ChatState *p_Chat;
extern LamportClock g_LamportClock;

/**
 * 如果表决模式（Voter_Working_Mode）是多个事务一次表决，需要注意表决的顺序。这时，
 * 表决时，下标为 cur_trans_id 的数据是最新的，所以从下标为 cur_trans_id+1 的数据开始
 * 中循环，因为 trans_arr 是循环递归使用的。
 *
 * 这就是这几行代码要由来：
 * int j = p_Chat->cur_trans_id;
 * j = (j + 1) % MAX_NUM_TRANS;
 */

/**
 * process_vote
 */
int process_vote(void)
{
    char     type;
    char    *dataRet = NULL;
    int32_t  dataLenRet = 0;
    Int64    i64;
    int      iRet;
    int      iShouldRet = MAX_NUM_TRANS;    // 如果待表决事务全部表决成功，则返回这个值

    i64.i64Val = 0;

    pthread_mutex_lock(&(p_Chat->mutex));

    if (p_Chat->numclients < g_pconfig->N_in_NooM) {
        type = VOTER_RESULT_FAIL;
    }
    else if (p_Chat->current_data_num < g_pconfig->N_in_NooM) {
        type = VOTER_RESULT_FAIL;
    }
    else {
        // Vote
        int j = p_Chat->cur_trans_id;
        VoterMessage *msg;

        for (int i = 0;  i < MAX_NUM_TRANS;  i++) {
            Transaction *trans = &(p_Chat->trans_arr[j]);

            if (VOTING_NULL != trans->votingResult) {
                continue;
            }

            msg = (VoterMessage *)trans->bigVotingData;

            j = (j + 1) % MAX_NUM_TRANS;

#ifdef __DBEUG__
            printf("process_vote (0), trans_idx: %d, "
                   "idTransaction: %ld, len: %d\n", j, trans->idTransaction, trans->lenData);
#endif
            int modeIdx = Vote(trans);
            if (0 != trans->isBigData) {
                if (modeIdx < 0) {
                    type = VOTER_RESULT_FAIL;
                }
                else {
                    type = VOTER_RESULT_RAW;
                    dataRet = (char*)(&(trans->bigVotingData[modeIdx]));
                    dataLenRet = trans->lenData;
                }
            }
            else {
                if (modeIdx < 0) {
                    type = VOTER_RESULT_FAIL;
                }
                else {
                    type = VOTER_RESULT_RAW;
                    i64 = trans->data[modeIdx];
                }
            }

            if (VOTER_RESULT_FAIL == type) {
                trans->votingResult = VOTING_NULL;
                if (0 != g_pconfig->StrongOrder) {
                    return 0;
                }
                else {
                    iShouldRet = 0;
                }
                continue;
            }

            // Pack the voting results data
            p_Chat->trans_arr[p_Chat->cur_trans_id].mode = i64;

            VoterMessage *message;
            char          messageBuf[MAX_VoterMessage_SIZE];
            uint64_t      clock;

            clock = local_event(&g_LamportClock);
            message = BoxVoterMessage(messageBuf,
                                            MAX_VoterMessage_SIZE,
                                            type,
                                            msg->idTransaction,
                                                        // uint64_t idTransaction
                                            clock,      // uint64_t lamportClock
                                            i64.i64Val, // int64_t data
                                            dataRet,    // char *rawData
                                            dataLenRet
                                        );

            // Send the voting results to all clients
            // SendMsgToAllClients(-1, (char*)message, message->len);

            trans->votingResult = VOTING_PASS;
            notify_select_exit(trans->idTransaction);

            g_voterOutputTokenOrder = 3;

#ifdef SYNC_WITH_PHXPAXOS
            if ((0 == g_pconfig->onlyMasterOutput) ||
                (('p' != g_pconfig->cDataSynchronizationProtocol) ||
                                        (0 == PhxIsIMMaster(g_oPaxosServer)))) {
#else
            if (true) {
#endif
                /**
                 * 一个表决器集群，由多个节点组成，这些表决节点必须多数表决通过才能算总的表决通
                 * 过，以此来防止节点失效，达到功能安全的目的。那种每个节点自己表决通过后，把表
                 * 决结果通过PhxPaxos同步出去，真要对外输出表决结果时，要判断是不是大多数的表
                 * 决器都表决通过。这中间要有一个延迟。
                 */

                int64_t delayTime;

                int32_t i = 0;
                do {
                    iRet = ClusterCheckVoterResult(message);
                    if (0 == iRet) {
                        break;
                    }
                    (void)TimespecWaitNanoseconds(NULL, &delayTime,
                                                  NANOSECONDS_FOR_SCHEDULE * 3);

                    i++;
                } while (i < g_pconfig->maxWait);

                if (i >= g_pconfig->maxWait) {
                    pthread_mutex_unlock(&(p_Chat->mutex));

                    // 表决未通过
                    syslog(LOG_ERR, "ClusterCheckVoterResult failed. iRet: %d", iRet);
                    return -1;
                }

                if (NULL != p_Chat->pVoterOutputMsg) {
                    iRet = p_Chat->pVoterOutputMsg(
                         (char*)message, message->len, 3, VOTER_TOKEN_3, trans);
                }
                else {
                    iRet = VoterOutputMsg(
                         (char*)message, message->len, 3, VOTER_TOKEN_3, trans);
                }

                if (iRet < 0) {
                    g_pVoterRunningState->votingResultFail++;
                    WriteLog("error, voting_voter_main", -1);
                }
            }
        } // for
    } // else

    p_Chat->cur_trans_id = 0;
    pthread_mutex_unlock(&(p_Chat->mutex));

    return iShouldRet;
}

/**
 * try to process_vote
 */
int try_to_process_vote(int justTry)
{
    char     type;
    char    *dataRet = NULL;
    int32_t  dataLenRet = 0;
    Int64    i64;
    int      iRet;
    int      iShouldRet = MAX_NUM_TRANS;    // 如果待表决事务全部表决成功，则返回这个值

    i64.i64Val = 0;

    if (p_Chat->numclients < g_pconfig->M_in_NooM) {
        type = VOTER_RESULT_FAIL;
    }
    else if (p_Chat->current_data_num < g_pconfig->N_in_NooM) {
        type = VOTER_RESULT_FAIL;
    }
    else {
        // Vote
        int j = p_Chat->cur_trans_id;
        VoterMessage *msg;

        for (int i = 0;  i < MAX_NUM_TRANS;  i++) {
            Transaction *trans = &(p_Chat->trans_arr[j]);

            if (VOTING_NULL != trans->votingResult) {
                continue;
            }

            msg = (VoterMessage *)trans->bigVotingData;

#ifdef __DBEUG__try_to_process_vote
            printf("try_to_process_vote (0), trans_idx: %d, "
                   "idTransaction: %ld, len: %d\n", j, trans->idTransaction, trans->lenData);
#endif
            j++;
            j = j % MAX_NUM_TRANS;

            int modeIdx = Vote(trans);
            if (0 != trans->isBigData) {
                if (modeIdx < 0) {
                    type = VOTER_RESULT_FAIL;
                }
                else {
                    type = VOTER_RESULT_RAW;
                    dataRet = (char*)(&(trans->bigVotingData[modeIdx]));
                    dataLenRet = trans->lenData;
                }
            }
            else {
                if (modeIdx < 0) {
                    type = VOTER_RESULT_FAIL;
                }
                else {
                    type = VOTER_RESULT_RAW;
                    i64 = trans->data[modeIdx];
                }
            }

            if (VOTER_RESULT_FAIL == type) {
                trans->votingResult = VOTING_NULL;
                if (0 != g_pconfig->StrongOrder) {
                    return 0;
                }
                else {
                    iShouldRet = 0;
                }
                continue;
            }

            if (0 != justTry) {
                continue;
            }

            // Pack the voting results data
            p_Chat->trans_arr[p_Chat->cur_trans_id].mode = i64;

            VoterMessage *message;
            char          messageBuf[MAX_VoterMessage_SIZE];
            uint64_t      clock;

            clock = local_event(&g_LamportClock);
            message = BoxVoterMessage(messageBuf,
                                            MAX_VoterMessage_SIZE,
                                            type,
                                            msg->idTransaction,
                                                        // uint64_t idTransaction
                                            clock,      // uint64_t lamportClock
                                            i64.i64Val, // int64_t data
                                            dataRet,    // char *rawData
                                            dataLenRet
                                        );

            // Send the voting results to all clients
            // SendMsgToAllClients(-1, (char*)message, message->len);

            trans->votingResult = VOTING_PASS;
            notify_select_exit(trans->idTransaction);

            g_voterOutputTokenOrder = 3;

#ifdef SYNC_WITH_PHXPAXOS
            if ((0 == g_pconfig->onlyMasterOutput) ||
                (('p' != g_pconfig->cDataSynchronizationProtocol) ||
                                        (0 == PhxIsIMMaster(g_oPaxosServer)))) {
#else
            if (true) {
#endif
                if (NULL != p_Chat->pVoterOutputMsg) {
                    iRet = p_Chat->pVoterOutputMsg((char*)message, message->len,
                                                       3, VOTER_TOKEN_3, trans);
                }
                else {
                    iRet = VoterOutputMsg((char*)message, message->len, 3,
                                                          VOTER_TOKEN_3, trans);
                }

                if (iRet < 0) {
                    g_pVoterRunningState->votingResultFail++;
                    WriteLog("error, voting_voter_main", -1);
                }
            }
        } // for
    } // else

    return iShouldRet;
}

/**
 * Cluster check VoterResult
 */
int ClusterCheckVoterResult(VoterMessage *msg)
{
#ifdef SYNC_WITH_PHXPAXOS
    int num = 0;
    int total = 0;

    int arch_count[MAX_ARCH_SUPPORT];
    memset(&arch_count, '\0', sizeof(arch_count));

    for (int i = 0;  i < p_Chat->cur_numVoters;  i++) {
        char     keybuf[MAX_ID_SIZE + 64];
        int16_t  vr = VOTING_FAIL;
        uint64_t llInstanceID;
        size_t   bufSize = sizeof(vr);

        snprintf(keybuf, sizeof(keybuf), MACRO_VR_PRINTF_FMTSTR,
                                       p_Chat->idVoters[i], msg->idTransaction);
        int iRet = PhxLevelDbRead(keybuf, (char *)&vr, &bufSize, &llInstanceID);
        if (0 == iRet) {
            total++;
            if (VOTING_PASS == vr) {
                num++;

                // 节点异构计算是否满足要求计数
                if (g_pconfig->arch_total > 0) {
                    for (int j = 0;  j < g_pconfig->arch_total;  j++) {
                        int match_len = 0;
                        int pos;
                        const char *p = p_Chat->idVoters[i];

                        while ((pos = re_matchp((re_t)&g_pconfig->archPatterns[j],
                                                          p, &match_len)) != -1) {
                            arch_count[j]++;
                            p += pos + match_len;
                        }
                    }
                }
            }
        }
    }

    // 节点异构计算是否满足要求判断
    if (g_pconfig->arch_total > 0) {
        for (int j = 0;  j < g_pconfig->arch_total;  j++) {
            if (arch_count[j] <= 0) {
                return 1;
            }
        }
    }
#ifdef __DBEUG__
        printf(">>>>>>>> ClusterCheckVoterResult(msg), "
                                 "total: %d VOTING_PASS num: %d\n", total, num);
#endif
    if ((total < g_pconfig->minimum_VOTER_WORKING) ||
                                       (num < g_pconfig->minimum_VOTING_PASS)) {
        return 1;
    }
#else
    (void)msg;
#endif

    return 0;
}
